package com.vote.services.impl;

import static me.prettyprint.cassandra.utils.StringUtils.string;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import me.prettyprint.cassandra.service.CassandraClient;
import me.prettyprint.cassandra.service.CassandraClientPool;
import me.prettyprint.cassandra.service.CassandraClientPoolFactory;
import me.prettyprint.cassandra.service.Keyspace;
import me.prettyprint.cassandra.service.PoolExhaustedException;

import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.SuperColumn;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;

import com.vote.model.Comment;
import com.vote.services.CommentService;

public class CommentServiceCassandraImpl implements CommentService {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(CommentServiceCassandraImpl.class);

    private CassandraClientPool pool = null;
    private CassandraClient client = null;
    private static final String ColumnFamily = "comment";
    private String host;
    private String port;
    private String keySpace;
    
    public void init(){
    	pool =  CassandraClientPoolFactory.INSTANCE.get();
    	try {
			client = pool.borrowClient(host, NumberUtils.toInt(port));
		} catch (IllegalStateException e) {
			logger.error("cassandra init fail~~", e);
		} catch (PoolExhaustedException e) {
			logger.error("cassandra init fail~~", e);
		} catch (Exception e) {
			logger.error("cassandra init fail~~", e);
		}
		logger.info("cassandra init success~~");
    }
    
    public void destroy(){
    }
	@Override
	public List<Comment> getCommentListByQuestionnaireId(int questionnaireId,
			int offset, int length) {
		
		List<SuperColumn> sclist = null;
		
		try {
			Keyspace keyspace = client.getKeyspace(keySpace);
			
			sclist = null;
			
			SlicePredicate sp = new SlicePredicate();
			
			SliceRange sr = new SliceRange();
			
			sr.setStart(new byte[0]);
			sr.setFinish(new byte[0]);
			sr.setReversed(true);
			
			sp.setSlice_range(sr);
			
			sclist = keyspace.getSuperSlice(Integer.toString(questionnaireId), new ColumnParent(ColumnFamily), sp);
		} catch (InvalidRequestException e) {
			e.printStackTrace();
		} catch (NotFoundException e) {
			e.printStackTrace();
		} catch (UnavailableException e) {
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
		} catch (TimedOutException e) {
			e.printStackTrace();
		}
		
		List<Comment> commentList = new ArrayList<Comment>();
		
		for(SuperColumn sc : sclist){
			Comment c = superColumn2comment(sc, questionnaireId);
			commentList.add(c);
		}
		
		return commentList;
	}

	@Override
	public int insertComment(Comment comment) throws Exception {

	    Keyspace keyspace = client.getKeyspace(keySpace);
	    
	    long time = System.currentTimeMillis();
	    
	    List<SuperColumn> cslist = new ArrayList<SuperColumn>();

	    cslist.add(comment2superColumn(comment, time));
	    
	    Map<String, List<SuperColumn>> cfmap = new HashMap<String, List<SuperColumn>>();
	    
	    cfmap.put(ColumnFamily, cslist);
	    
	    keyspace.batchInsert(Integer.toString(comment.getQuestionnaireId()), null, cfmap);
	    
		return 0;
	}
	
	private SuperColumn comment2superColumn(Comment comment, long time){
		List<Column> cl = new ArrayList<Column>();
		Column fromUserId = new Column("fromUserId".getBytes(), Integer.toString(comment.getFromUserId()).getBytes(), time);
		cl.add(fromUserId);
		Column toUserId = new Column("toUserId".getBytes(), Integer.toString(comment.getToUserId()).getBytes(), time);
		cl.add(toUserId);
		Column commentText = new Column("commentText".getBytes(), comment.getCommentText().getBytes(), time);
		cl.add(commentText);
	    SuperColumn sc = new SuperColumn();
	    sc.setColumns(cl);
	    
	    sc.setName(Long.toString(time).getBytes());
		return sc;
	}
	
	private Comment superColumn2comment(SuperColumn sc, int questionnaireId){
		
		Comment comment = new Comment();
		
		comment.setQuestionnaireId(questionnaireId);
		comment.setCommentTime(new Timestamp(NumberUtils.toLong(string(sc.getName()))));
		
		for(Column c : sc.columns){
			if("fromUserId".equalsIgnoreCase(string(c.getName()))){
				comment.setFromUserId(NumberUtils.toInt(string(c.getValue())));
			}
			else if("toUserId".equalsIgnoreCase(string(c.getName()))){
				comment.setToUserId(NumberUtils.toInt(string(c.getValue())));
			}
			else if("commentText".equalsIgnoreCase(string(c.getName()))){
				comment.setCommentText(string(c.getValue()));
			}
		}
		
		return comment;
	}
	public String getHost() {
		return host;
	}
	public void setHost(String host) {
		this.host = host;
	}
	public String getPort() {
		return port;
	}
	public void setPort(String port) {
		this.port = port;
	}
	public String getKeySpace() {
		return keySpace;
	}
	public void setKeySpace(String keySpace) {
		this.keySpace = keySpace;
	}

	@Override
	public int getCommentListCountByQuestionnaireId(int questionnaireId) {
		// TODO Auto-generated method stub
		return 0;
	}
	
}
